/*******************************************************************************
 * * Copyright 2011 Impetus Infotech.
 *  *
 *  * Licensed under the Apache License, Version 2.0 (the "License");
 *  * you may not use this file except in compliance with the License.
 *  * You may obtain a copy of the License at
 *  *
 *  *      http://www.apache.org/licenses/LICENSE-2.0
 *  *
 *  * Unless required by applicable law or agreed to in writing, software
 *  * distributed under the License is distributed on an "AS IS" BASIS,
 *  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  * See the License for the specific language governing permissions and
 *  * limitations under the License.
 ******************************************************************************/
package com.scalyze.apps.socialanyltics.datastore.persistence;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.util.Bytes;
import org.json.JSONObject;

/**
 * @author impetus
 */
public class HBaseDataHandler implements DataHandler
{
    /** the log used by this class. */
    // private static Log log = LogFactory.getLog(HBaseDataHandler.class);

    private HBaseConfiguration conf;

    private HBaseAdmin admin;

    private Writer hbaseWriter = new HBaseWriter();

    public HBaseDataHandler(String hostName, String port,String zookeeperHost)
    {
        try
        {
            init(hostName, port, zookeeperHost);
        }
        catch (MasterNotRunningException e)
        {
            throw new RuntimeException(e.getMessage());
        }
    }

    // @Override
    private synchronized void createTableIfDoesNotExist(final String tableName, final Iterator<String> colFamily)
            throws MasterNotRunningException, IOException
    {
        HTableDescriptor htDescriptor = null;
        byte[] table = Bytes.toBytes(tableName);
        if (!admin.tableExists(table))
        {
            htDescriptor = new HTableDescriptor(tableName);
            addAndValidateColFamily(colFamily, htDescriptor, false);
            admin.createTable(htDescriptor);
            // return htDescriptor.getName();
        }
        else
        {
            htDescriptor = admin.getTableDescriptor(table);
            if(addAndValidateColFamily(colFamily, htDescriptor, true))
            {
                admin.disableTable(table);
                admin.modifyTable(table, htDescriptor);
                admin.enableTable(table);
            }
        }

    }

    private boolean  addAndValidateColFamily(final Iterator<String> colFamily, HTableDescriptor htDescriptor,
            boolean validate)
    {
        boolean newColumnFamilyAdded=false;
        while (colFamily.hasNext())
        {
            String familyName = colFamily.next();
            if (!validate || (validate && htDescriptor.getFamily(Bytes.toBytes(familyName)) == null))
            {
                HColumnDescriptor familyMetadata = new HColumnDescriptor(familyName);
                htDescriptor.addFamily(familyMetadata);
                newColumnFamilyAdded=true;
            }
        }
        return newColumnFamilyAdded;
    }

    @Override
    public void writeData(JSONObject object, String trendName) throws IOException
    {
        hbaseWriter.writeColumns(gethTable("HB" + trendName, object.keys()), object);

    }

    private void loadConfiguration(final String hostName, final String port, String zookeeperHost) throws MasterNotRunningException
    {
        Configuration hadoopConf = new Configuration();
        hadoopConf.set("hbase.master", hostName + ":" + port);
        
        conf = new HBaseConfiguration(hadoopConf);
        conf.set("hbase.zookeeper.quorum", zookeeperHost);
        getHBaseAdmin();
    }

    /**
     * 
     * @param hostName
     * @param port
     * @param zookeeperHost 
     * @throws MasterNotRunningException
     */
    private void init(final String hostName, final String port, String zookeeperHost) throws MasterNotRunningException
    {
        if (conf == null)
        {
            loadConfiguration(hostName, port, zookeeperHost);
        }
    }

    /**
     * @throws MasterNotRunningException
     */
    private void getHBaseAdmin() throws MasterNotRunningException
    {
        try
        {
            admin = new HBaseAdmin(conf);
        }
        catch (ZooKeeperConnectionException e)
        {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    private HTable gethTable(final String tableName, Iterator<String> colFamilies) throws IOException
    {
        createTableIfDoesNotExist(tableName, colFamilies);
        return new HTable(conf, tableName);
    }

    @Override
    public void shutdown()
    {
        try
        {
            admin.shutdown();
        }
        catch (IOException e)
        {
            throw new RuntimeException(e.getMessage());
        }
    }

    public static void main(String[] args)
    {
        HBaseDataHandler handler = new HBaseDataHandler("127.0.0.1", "60000", "127.0.0.1");
    }
}
